package com.xbk.seckill.order.application.place.impl;

import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.NumberUtil;
import com.alibaba.fastjson.JSONObject;
import com.xbk.seckill.common.cache.distribute.DistributedCacheService;
import com.xbk.seckill.common.constants.SeckillConstants;
import com.xbk.seckill.common.exception.ErrorCode;
import com.xbk.seckill.common.exception.SeckillException;
import com.xbk.seckill.common.model.dto.SeckillGoodsDTO;
import com.xbk.seckill.common.model.message.TxMessage;
import com.xbk.seckill.common.utils.id.SnowFlakeFactory;
import com.xbk.seckill.dubbo.interfaces.goods.SeckillGoodsDubboService;
import com.xbk.seckill.mq.MessageSenderService;
import com.xbk.seckill.order.application.model.command.SeckillOrderCommand;
import com.xbk.seckill.order.application.place.SeckillPlaceOrderService;
import com.xbk.seckill.order.domain.model.entity.SeckillOrder;
import com.xbk.seckill.order.domain.service.SeckillOrderDomainService;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.TimeUnit;

/**
 * 基于lua脚本防止库存超卖
 *
 * @author xiexu
 * @date 2025/02/07 10:54
 */
@Slf4j
@Service
@ConditionalOnProperty(name = "place.order.type", havingValue = "lua")
public class SeckillPlaceOrderLuaService implements SeckillPlaceOrderService {

    @Autowired
    private SeckillOrderDomainService seckillOrderDomainService;

    @DubboReference(version = "1.0.0", check = false)
    private SeckillGoodsDubboService seckillGoodsDubboService;

    @Autowired
    private DistributedCacheService distributedCacheService;

    @Autowired
    private MessageSenderService messageSenderService;

    /**
     * 下单操作
     *
     * @param userId
     * @param seckillOrderCommand
     * @return
     */
    @Override
    public Long placeOrder(Long userId, SeckillOrderCommand seckillOrderCommand) {
        // 根据商品id和版本号，获取商品详细信息
        SeckillGoodsDTO seckillGoods = seckillGoodsDubboService.getSeckillGoods(seckillOrderCommand.getGoodsId(), seckillOrderCommand.getVersion());
        // 校验商品信息
        checkSeckillGoods(seckillOrderCommand, seckillGoods);
        boolean exception = false;
        // 通过雪花算法，生成一个全局唯一的事务id
        long txNo = SnowFlakeFactory.getSnowFlakeFromCache().nextId();
        // 商品库存key
        String key = SeckillConstants.getKey(SeckillConstants.GOODS_ITEM_STOCK_KEY_PREFIX, String.valueOf(seckillOrderCommand.getGoodsId()));
        Long decrementResult = 0L;
        try {
            // 获取商品限购信息
            String limitKey = SeckillConstants.getKey(SeckillConstants.GOODS_ITEM_LIMIT_KEY_PREFIX, String.valueOf(seckillOrderCommand.getGoodsId()));
            Object limitObj = distributedCacheService.getObject(limitKey);
            // 如果从Redis获取到的限购信息为null，则说明商品已经下线
            if (limitObj == null) {
                throw new SeckillException(ErrorCode.GOODS_OFFLINE);
            }
            // 校验限购数量
            if (Integer.parseInt(String.valueOf(limitObj)) < seckillOrderCommand.getQuantity()) {
                throw new SeckillException(ErrorCode.BEYOND_LIMIT_NUM);
            }
            // 使用lua脚本扣减缓存库存
            decrementResult = distributedCacheService.decrementByLua(key, seckillOrderCommand.getQuantity());
            // 校验lua脚本扣减的结果
            this.checkResult(decrementResult);
        } catch (Exception e) {
            log.error("SeckillPlaceOrderLuaService|下单异常|参数:{}|异常信息:{}", JSONObject.toJSONString(seckillOrderCommand), e.getMessage());
            exception = true;
            if (decrementResult == SeckillConstants.LUA_RESULT_EXECUTE_TOKEN_SUCCESS) {
                // 将缓存中的库存增加回去
                distributedCacheService.incrementByLua(key, seckillOrderCommand.getQuantity());
            }
        }
        // 发送事务消息
        TxMessage message = this.getTxMessage(SeckillConstants.TOPIC_TX_MSG, txNo, userId, SeckillConstants.PLACE_ORDER_TYPE_LUA, exception, seckillOrderCommand, seckillGoods, 0, seckillOrderCommand.getOrderTaskId());
        messageSenderService.sendMessageInTransaction(message, null);
        return txNo;
    }

    /**
     * 本地事务执行保存订单操作
     *
     * @param txMessage
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveOrderInTransaction(TxMessage txMessage) {
        // 拼接订单事务key
        String orderTxKey = SeckillConstants.getKey(SeckillConstants.ORDER_TX_KEY, String.valueOf(txMessage.getTxNo()));
        try {
            // 幂等判断
            Boolean submitTransaction = distributedCacheService.hasKey(orderTxKey);
            if (BooleanUtil.isTrue(submitTransaction)) {
                log.info("saveOrderInTransaction|已经执行过本地事务|{}", txMessage.getTxNo());
                return;
            }
            // 构建订单对象
            SeckillOrder seckillOrder = this.buildSeckillOrder(txMessage);
            // 保存订单到数据库
            seckillOrderDomainService.saveSeckillOrder(seckillOrder);
            // 保存事务日志
            distributedCacheService.put(orderTxKey, txMessage.getTxNo(), SeckillConstants.TX_LOG_EXPIRE_DAY, TimeUnit.DAYS);
        } catch (Exception e) {
            log.error("saveOrderInTransaction|异常|{}", e.getMessage());
            // 删除事务日志
            distributedCacheService.delete(orderTxKey);
            // 回滚缓存库存
            this.rollbackCacheStack(txMessage);
            throw e;
        }
    }

    /**
     * 回滚缓存库存
     */
    private void rollbackCacheStack(TxMessage txMessage) {
        // txMessage的exception字段为false，说明前面扣减过缓存库存
        if (BooleanUtil.isFalse(txMessage.getException())) {
            String luaKey = SeckillConstants.getKey(SeckillConstants.ORDER_TX_KEY, String.valueOf(txMessage.getTxNo())).concat(SeckillConstants.LUA_SUFFIX);
            Long result = distributedCacheService.checkExecute(luaKey, SeckillConstants.TX_LOG_EXPIRE_SECONDS);
            // 已经执行过恢复缓存库存的方法
            if (NumberUtil.equals(result, SeckillConstants.CHECK_RECOVER_STOCK_HAS_EXECUTE)) {
                log.info("handlerCacheStock|已经执行过恢复缓存库存的方法|{}", JSONObject.toJSONString(txMessage));
                return;
            }
            // 只有分布式锁方式和Lua脚本方法才会扣减缓存中的库存
            String key = SeckillConstants.getKey(SeckillConstants.GOODS_ITEM_STOCK_KEY_PREFIX, String.valueOf(txMessage.getGoodsId()));
            distributedCacheService.increment(key, txMessage.getQuantity());
        }
    }

    /**
     * 校验lua脚本执行的结果
     *
     * @param result
     */
    private void checkResult(Long result) {
        if (result == SeckillConstants.LUA_RESULT_GOODS_STOCK_NOT_EXISTS) {
            throw new SeckillException(ErrorCode.STOCK_IS_NULL);
        }
        if (result == SeckillConstants.LUA_RESULT_GOODS_STOCK_PARAMS_LT_ZERO) {
            throw new SeckillException(ErrorCode.PARAMS_INVALID);
        }
        if (result == SeckillConstants.LUA_RESULT_GOODS_STOCK_LT_ZERO) {
            throw new SeckillException(ErrorCode.STOCK_LT_ZERO);
        }
    }

}
